package com.gitee.ixtf.vertx.verticle

import com.gitee.ixtf.vertx.kotlinx.failure
import io.vertx.core.eventbus.Message
import io.vertx.kotlin.core.eventbus.deliveryOptionsOf
import io.vertx.kotlin.coroutines.awaitResult
import io.vertx.kotlin.coroutines.coAwait

abstract class QueueConsumerCoroutineVerticle<T>(address: String) :
    ConsumerCoroutineVerticle<T, Boolean>(address) {
  private val queueConsumer by lazy { eventBus.consumer<T>("$address-queue").pause().fetch(1) }
  private var totalCount: Long = 0
  private var successCount: Long = 0
  private var failureCount: Long = 0

  protected abstract suspend fun handleQueue(reply: Message<T>)

  protected open suspend fun handleQueueSuccess(reply: Message<T>) = Unit

  protected open suspend fun handleQueueFailure(reply: Message<T>, t: Throwable) =
      reply.failure(log, t)

  protected open suspend fun handleQueueFinally(reply: Message<T>) = Unit

  override suspend fun start() {
    queueConsumer.coHandler { reply ->
      runCatching { handleQueue(reply) }
          .onSuccess {
            successCount++
            handleQueueSuccess(reply)
          }
          .onFailure {
            failureCount++
            handleQueueFailure(reply, it)
          }
      queueConsumer.fetch(1)
      handleQueueFinally(reply)
    }
    awaitResult { queueConsumer.completionHandler(it) }
    super.start()
  }

  override suspend fun stop() {
    super.stop()
    queueConsumer.unregister().coAwait()
  }

  final override suspend fun handle(reply: Message<T>): Boolean {
    val deliveryOptions = deliveryOptionsOf()
    reply.headers().forEach(deliveryOptions::addHeader)
    eventBus.send(queueConsumer.address(), reply.body(), deliveryOptions)
    totalCount++
    return true
  }

  final override suspend fun handleSuccess(reply: Message<T>, r: Boolean) =
      super.handleSuccess(reply, r)

  final override suspend fun handleFailure(reply: Message<T>, t: Throwable) =
      super.handleFailure(reply, t)

  final override suspend fun handleFinally(reply: Message<T>) = super.handleFinally(reply)
}
